package cn.itcast.dtx.txmsgdemo.bank1.message;

import cn.itcast.dtx.txmsgdemo.bank1.dao.AccountInfoDao;
import cn.itcast.dtx.txmsgdemo.bank1.model.AccountChangeEvent;
import cn.itcast.dtx.txmsgdemo.bank1.service.AccountInfoService;
import com.fasterxml.jackson.databind.ObjectMapper;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.spring.annotation.RocketMQTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionListener;
import org.apache.rocketmq.spring.core.RocketMQLocalTransactionState;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.messaging.Message;
import org.springframework.stereotype.Component;
import org.springframework.transaction.annotation.Transactional;

import java.util.Map;

/**
 * @author zh_o
 * @date 2020-11-14
 **/
@Slf4j
@Component
@RocketMQTransactionListener
public class ProducerTransactionMsgListener implements RocketMQLocalTransactionListener {

    @Autowired
    private AccountInfoService accountInfoService;

    @Autowired
    private ObjectMapper json;

    @Autowired
    private AccountInfoDao accountInfoDao;

    /**
     * 事务消息发送后的回调方法，当消息发送给 mq 成功，此方法被回调
     * @param message 消息
     * @param o 参数
     */
    @Override
    @Transactional(rollbackFor = Exception.class)
    public RocketMQLocalTransactionState executeLocalTransaction(Message message, Object o) {
        try {
            //解析 message，转为 AccountChangeEvent
            String jsonStr = json.writeValueAsString(message.getPayload());
            Map<String, Object> map = json.readValue(jsonStr, Map.class);
            Object obj = map.get("accountChange");
            AccountChangeEvent accountChangeEvent = json.readValue(json.writeValueAsString(obj), AccountChangeEvent.class);
            // 执行本地事务，扣减金额
            accountInfoService.doUpdateAccountBalance(accountChangeEvent);
            // 本地事务正常，通知消息服务提交，消息服务将消息状态改为可消费
            return RocketMQLocalTransactionState.COMMIT;
        } catch (Exception e) {
            e.printStackTrace();
            // 本地事务异常，通知消息服务回滚删除消息
            return RocketMQLocalTransactionState.ROLLBACK;
        }


    }

    /**
     * 事务状态回查，查询是否扣减金额
     * @param message 消息
     */
    @SneakyThrows
    @Override
    public RocketMQLocalTransactionState checkLocalTransaction(Message message) {
        //解析 message，转成 AccountChangeEvent
        String jsonStr = json.writeValueAsString(message.getPayload());
        Map<String, Object> map = json.readValue(jsonStr, Map.class);
        Object obj = map.get("accountChange");
        AccountChangeEvent accountChangeEvent = json.readValue(json.writeValueAsString(obj), AccountChangeEvent.class);
        // 获取事务ID
        String txNo = accountChangeEvent.getTxNo();
        // 判断事务是否执行成功
        int existTx = accountInfoDao.isExistTx(txNo);
        if (existTx > 0){
            // 扣款成功
            return RocketMQLocalTransactionState.COMMIT;
        } else {
            // 扣款失败
            return RocketMQLocalTransactionState.UNKNOWN;
        }
    }

}
